package com.fanxl.flink.project.app;

import com.fanxl.flink.project.udf.AccessKafkaDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.connector.kafka.source.reader.deserializer.KafkaRecordDeserializationSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import java.io.IOException;

/**
 * @description:
 * @author: fanxiaole
 * @date: 2022/3/20 22:45
 */
public class KafkaApp2 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        KafkaSource<Tuple2<String, String>> source = KafkaSource.<Tuple2<String, String>>builder()
                .setBootstrapServers("10.215.7.21:9092,10.215.7.23:9092,10.215.7.24:9092")
                .setTopics("flink_test")
                .setGroupId("test_group")
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setDeserializer(new AccessKafkaDeserializationSchema())
                .setProperty("enable.auto.commit", "false")
                .build();

        DataStreamSource<Tuple2<String, String>> streamSource = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");
        streamSource.print();
        env.execute("KafkaApp2");
    }
}
